
package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.date.BetweenFormater;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.PrimitiveArrayUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hexinfo.dmpro.common.model.ClusterSparingConf;
import com.hexinfo.dmpro.common.model.MessageCenter;
import com.hexinfo.dmpro.common.service.MessageCenterService;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.common.utils.StringUtils;
import com.hexinfo.dmpro.sparing.dao.SyncRuleConfMapper;
import com.hexinfo.dmpro.sparing.dto.*;
import com.hexinfo.dmpro.sparing.model.*;
import com.hexinfo.dmpro.sparing.service.*;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * 万国到UAT同步 (源级)
 *
 * @author yemw
 * @date 2023-09-08 17:45:29
 */
@Service
@AllArgsConstructor
@Slf4j
public class SyncRuleConfServiceTWOImpl extends ServiceImpl<SyncRuleConfMapper, SyncRuleConf> implements SyncRuleConfTWOService {

    private final ScanMetadataService scanMetadataService;

    private final TimeFieldMatchingService timeFieldMatchingService;

    private final HiveToOracleService hiveToOracleService;

    private final RuleToDistCPService ruleToDistCPService;

    private final MessageCenterService messageCenterService;

    private final AvailabilityPathService availabilityPathService;

    private final UatSyncLogsTaskService uatSyncLogsTaskService;

    private final UatSyncLogsService uatSyncLogsService;

    private SyncRuleConfServiceImpl syncRuleConfServiceImpl;


    @Override
    public UATSyncSearchDTO uATSyncSearch(HandMoveSyncDTO handMoveSyncDTO) {
        UATSyncSearchDTO dto = new UATSyncSearchDTO();
        /**
         * 源名集合
         */
        List<String> sourceName = new ArrayList<>();
        /**
         * 库名集合
         */
        List<String> tblDatabaseNameList = new ArrayList<>();
        /**
         * 表名集合
         */
        List<String> tblNameList = new ArrayList<>();
        dto.setSourceName(sourceName);
        dto.setTblDatabaseNameList(tblDatabaseNameList);
        dto.setTblNameList(tblNameList);
        //源信息为空 查询源信息
        if(StrUtil.isBlank(handMoveSyncDTO.getSource())){
            LambdaQueryWrapper<SyncRuleConf> syncRuleWrapper = Wrappers.lambdaQuery();
            syncRuleWrapper.select(SyncRuleConf::getSourceName).eq(SyncRuleConf::getDel, CommonConstants.STATUS_NORMAL).orderByDesc(SyncRuleConf::getSourceName).groupBy(SyncRuleConf::getSourceName);
            List<Object> syncRuleConfs = this.baseMapper.selectObjs(syncRuleWrapper);
            for (Object syncRuleConf: syncRuleConfs) {
                if(ObjectUtil.isEmpty(syncRuleConf)){
                    dto.getSourceName().add("无源");
                }else{
                    dto.getSourceName().add(syncRuleConf.toString());
                }
            }
        }else{
            //库信息为空 查询该源的库信息 (数据库分组查询数据量过大导致查询过慢 所以使用静态值)
            if(StrUtil.isBlank(handMoveSyncDTO.getDbName())){
                dto.getTblDatabaseNameList().add("ods");
                dto.getTblDatabaseNameList().add("dw");
                dto.getTblDatabaseNameList().add("stg");
            }else{
                //查询规则
                LambdaQueryWrapper<SyncRuleConf> syncRuleWrapper = Wrappers.lambdaQuery();
                syncRuleWrapper.eq(SyncRuleConf::getDel, CommonConstants.STATUS_NORMAL)
                        .eq(StrUtil.isNotBlank(handMoveSyncDTO.getSource())&&!"无源".equals(handMoveSyncDTO.getSource()),SyncRuleConf::getSourceName,handMoveSyncDTO.getSource())
                        .isNull(StrUtil.isBlank(handMoveSyncDTO.getSource())||"无源".equals(handMoveSyncDTO.getSource()),SyncRuleConf::getSourceName);
                List<SyncRuleConf> syncRuleConfs = this.baseMapper.selectList(syncRuleWrapper);
                syncRuleConfs.forEach(syncRule -> {
                    ScanSyncRuleDTO syncRuleDTO = syncRuleConfServiceImpl.splQuSoeDataTable(syncRule);
                    syncRuleDTO.setTblDatabaseName(handMoveSyncDTO.getDbName());
                    List<ScanMetadata> scanMetadataList = scanMetadataService.qyScanMetadataTable(syncRuleDTO);
                    for(ScanMetadata scanMetadata : scanMetadataList){
                        dto.getTblNameList().add(scanMetadata.getTblName());
                    }
                });
            }
        }
        return dto;
    }

    /**
     * 特殊规则黑名单校验（时间分区）
     * @param sourceName
     * @return
     */
    @Override
    public Map<String,String> upTableBameBlack(String sourceName) {
        Map<String,String> map = new HashMap<>();
        QueryWrapper<SyncRuleConf>  wrapperGO  = new QueryWrapper<>();
        wrapperGO.select(" NVL(source_name, '无源') as source_name ").lambda()
                .eq(StrUtil.isNotBlank(sourceName)&&!"无源".equals(sourceName),SyncRuleConf::getSourceName,sourceName)
                .isNull("无源".equals(sourceName),SyncRuleConf::getSourceName)
                .groupBy(SyncRuleConf::getSourceName);
        List<SyncRuleConf> syncRuleConfGO = this.baseMapper.selectList(wrapperGO);
        if(ObjectUtil.isNotEmpty(syncRuleConfGO)){
            for (SyncRuleConf syGO: syncRuleConfGO) {
                //查询当前特殊规则
                QueryWrapper<SyncRuleConf>  wrapper  = new QueryWrapper<>();
                wrapper.select("DISTINCT TABLE_NAME_WHITE").lambda()
                        .eq(StrUtil.isNotBlank(syGO.getSourceName())&&!"无源".equals(syGO.getSourceName()),SyncRuleConf::getSourceName,syGO.getSourceName())
                        .isNull("无源".equals(syGO.getSourceName()),SyncRuleConf::getSourceName)
                        .eq(SyncRuleConf::getIsTimePart,"特殊表含时间分区")
                        .eq(SyncRuleConf::getDel,CommonConstants.STATUS_NORMAL);
                List<SyncRuleConf> syncRuleConfs = this.baseMapper.selectList(wrapper);
                if(ObjectUtil.isNotEmpty(syncRuleConfs)){
                    for (SyncRuleConf sy: syncRuleConfs) {
                        //查询非特殊规则中黑名单中是否有改表名称
                        LambdaQueryWrapper<SyncRuleConf> wrapperA = Wrappers.lambdaQuery();
                        wrapperA.eq(StrUtil.isNotBlank(syGO.getSourceName())&&!"无源".equals(syGO.getSourceName()),SyncRuleConf::getSourceName,syGO.getSourceName())
                                .isNull("无源".equals(syGO.getSourceName()),SyncRuleConf::getSourceName)
                                .ne(SyncRuleConf::getIsTimePart,"特殊表含时间分区")
                                .ne(SyncRuleConf::getIsPart,"非分区")
                                .like(SyncRuleConf::getTableNameBlack,sy.getTableNameWhite())
                                .eq(SyncRuleConf::getDel,CommonConstants.STATUS_NORMAL);
                        List<SyncRuleConf> syncRuleConfsA = this.baseMapper.selectList(wrapperA);
                        //当前规则的黑名单不包含特殊表名
                        if(ObjectUtil.isEmpty(syncRuleConfsA)){
                            //修改当前规则
                            LambdaQueryWrapper<SyncRuleConf> wrapperB = Wrappers.lambdaQuery();
                            wrapperB.eq(StrUtil.isNotBlank(syGO.getSourceName())&&!"无源".equals(syGO.getSourceName()),SyncRuleConf::getSourceName,syGO.getSourceName())
                                    .isNull("无源".equals(syGO.getSourceName()),SyncRuleConf::getSourceName)
                                    .ne(SyncRuleConf::getIsTimePart,"特殊表含时间分区")
                                    .ne(SyncRuleConf::getIsTimePart,"无时间分区")
                                    .eq(SyncRuleConf::getDel,CommonConstants.STATUS_NORMAL);
                            List<SyncRuleConf> syncRuleConfsB = this.baseMapper.selectList(wrapperB);
                            if(ObjectUtil.isEmpty(syncRuleConfsB)){
                                for (SyncRuleConf syB: syncRuleConfsB) {
                                    syB.setTableNameBlack(syB.getBaseNameBlack()+","+sy.getTableNameWhite());
//                                    this.baseMapper.updateById(syB);
                                    map.put(syB.getSyncRuleName(),syB.getSyncRuleName());
                                }
                            }
                        }
                    }
                }
            }
        }
        return map;
    }



    /**
     * 上线测试
     * @return
     */
    @Override
    public Map<String,String> onlineTesting(TbHdfsDistcpDTO dto) {
        Map<String,String> map = new HashMap<>();
        //消息中心
        MessageCenter mess = new MessageCenter();
        mess.setMenu(ScanCommonConstants.ClusterName.SJRB.value);
        mess.setOperator("admin");
        mess.setDataType(ScanCommonConstants.ClusterName.JXZ.value);
        mess.setMessageSubject(ScanCommonConstants.ClusterName.SDSJTB.value);
        mess.setMessageWeight(ScanCommonConstants.ClusterName.ONE.value);
        messageCenterService.save(mess);
        //查询集群信息
        LambdaQueryWrapper<ClusterSparingConf> jQWrapper = Wrappers.lambdaQuery();
        LambdaQueryWrapper<ClusterSparingConf> wGWrapper = Wrappers.lambdaQuery();
        ClusterSparingConf jQCluster = scanMetadataService.wrapperClusterSparingConf(jQWrapper,"金桥");
        ClusterSparingConf wGCluster = scanMetadataService.wrapperClusterSparingConf(wGWrapper,"万国");
        if(ObjectUtil.isEmpty(jQCluster)){
            map.put("异常(金桥集群信息)","数据库表 TD_CLUSTER_SPARING_CONF 中未配置金桥集群信息");
        }else{
            map.put("金桥集群信息HDFS地址前缀  ",jQCluster.getHdfsPrefixPath());
        }
        if(ObjectUtil.isEmpty(wGCluster)){
            map.put("异常(万国集群信息)","数据库表 TD_CLUSTER_SPARING_CONF 中未配置万国集群信息");
        }else{
            map.put("万国集群信息HDFS地址前缀  ",wGCluster.getHdfsPrefixPath());
        }
        //查询消息中心信息
        LambdaQueryWrapper<MessageCenter> MCWrapper = Wrappers.lambdaQuery();
        MCWrapper.eq(MessageCenter::getDel,CommonConstants.STATUS_NORMAL)
                .eq(MessageCenter::getDataType,ScanCommonConstants.ClusterName.YWC.value)
                .like(MessageCenter::getMessageSubject,ScanCommonConstants.ClusterName.YSJHQ.value)
        ;
        List<MessageCenter> messageCenters = messageCenterService.list(MCWrapper);
        if(ObjectUtil.isEmpty(messageCenters)){
            map.put("异常(消息中心首次同步数据异常)","数据库表 TD_MESSAGE_CENTER 中未配置首次同步数据信息");
        }else{
            if(messageCenters.size()==18){
                map.put("消息中心首次同步数据源  ",messageCenters.size()+"个");
                map.put("消息中心同步时间为  ",""+messageCenters.get(0).getCreateTime());
            }else{
                map.put("异常(消息中心首次同步数据信息数量)","数据库表 TD_MESSAGE_CENTER 中首次同步数据信息数量异常");
            }
        }
        try {
            int a = hiveToOracleService.selectHiveTableCount();
            if(a>0){
                map.put("hive元数据库","连接成功");
            }
        } catch (Exception e) {
            map.put("异常(hive元数据库连接异常)","检查配置文件 /opt/dmpro/config/config/db.setting");
        }
        List<TbHdfsDistcpDTO> distcpDTOList = new ArrayList<>();
        distcpDTOList.add(dto);
        String s = ruleToDistCPService.hdfsDistcp(distcpDTOList, DateUtil.format(new Date(), "yyyyMMddHHmmss"),mess.getId());
        if(StrUtil.isNotBlank(s)){
            map.put("测试案例","异常");
        }else{
            map.put("测试案例","成功");
        }
        return map;
    }

    /**
     * 测试同步 UAT
     * @param dto
     * @return
     */
    @Override
    public Boolean handMoveSync(HandMoveSyncDTO dto) {
        String nameNodeStatus = availabilityPathService.getSouAndTar();
        //转小写
        dto.setDbName(StringUtils.toLowerCase(dto.getDbName()));
        dto.setTableName(StringUtils.toLowerCase(dto.getTableName()));
        if (StrUtil.isBlank(nameNodeStatus)){
            log.warn("-------------------高可用节点变动异常");
            return false;
        }
        Map<String,String> map = new HashMap<>();
        //消息中心
        MessageCenter mess = new MessageCenter();
        mess.setMenu(ScanCommonConstants.ClusterName.UATRB.value);
        mess.setOperator("admin");
        mess.setDataType(ScanCommonConstants.ClusterName.JXZ.value);
        mess.setMessageSubject(ScanCommonConstants.ClusterName.CSSJTB.value);
        mess.setMessageWeight(ScanCommonConstants.ClusterName.ONE.value);
        messageCenterService.save(mess);
        //任务表
        UatSyncLogsTask uatSyncLogsTask = new UatSyncLogsTask();
        //当前开始时间
        Date staDate = DateUtil.date();
        if(StrUtil.isNotBlank(dto.getSource())){
            uatSyncLogsTask.setTaskName(StringUtils.toUpperCase(dto.getSource())+DateUtil.format(new Date(), "yyyyMMdd"));
            uatSyncLogsTask.setSourceName("【源级】【增量】【"+dto.getStaDate()+"--"+dto.getEndDate()+"】");
        }else{
            uatSyncLogsTask.setTaskName("时间窗口"+DateUtil.format(new Date(), "yyyyMMdd"));
            uatSyncLogsTask.setSourceName("【源级】【增量】【"+dto.getStaDate()+"--"+dto.getEndDate()+"】");
        }
        uatSyncLogsTask.setTaskState(ScanCommonConstants.ClusterName.JXZ.value);
        uatSyncLogsTaskService.save(uatSyncLogsTask);
        //手动同步库标识
        String prefix = "uat_";
        //查询规则
        LambdaQueryWrapper<SyncRuleConf> syncRuleWrapper = Wrappers.lambdaQuery();
        syncRuleWrapper.eq(SyncRuleConf::getDel, CommonConstants.STATUS_NORMAL)
                .eq(StrUtil.isNotBlank(dto.getSource())&&!"dmpro".equals(dto.getSource()),SyncRuleConf::getSourceName,dto.getSource())
                .isNull(StrUtil.isBlank(dto.getSource()),SyncRuleConf::getSourceName);
        List<SyncRuleConf> syncRuleConfs = this.baseMapper.selectList(syncRuleWrapper);
        //查询集群信息
        LambdaQueryWrapper<ClusterSparingConf> wGWrapper = Wrappers.lambdaQuery();
        LambdaQueryWrapper<ClusterSparingConf> uatWrapper = Wrappers.lambdaQuery();
        ClusterSparingConf wGCluster = scanMetadataService.wrapperClusterSparingConf(wGWrapper,"万国");
        ClusterSparingConf uatCluster = scanMetadataService.wrapperClusterSparingConf(uatWrapper,"UAT");
        //DistCP集合
        List<HdfsHeatBackup> hdfsHeatBackups = new ArrayList<>();
        //对比条件集合
        List<WhereScriptDTO> whereScriptDTOS = new ArrayList<>();
        RuleToDistCPDTO ruleToDistCPDTO = new RuleToDistCPDTO();
        syncRuleConfs.forEach(syncRule -> {
            ScanSyncRuleDTO syncRuleDTO = syncRuleConfServiceImpl.splQuSoeDataTable(syncRule);
            syncRuleDTO.setTblDatabaseName(dto.getDbName());
            syncRuleDTO.setTblName(dto.getTableName());
            //判断同步范围 全量还是近2年 0为全量
            //全量数据不需要判断时间分区
            if(syncRule.getSyncRange()==0){
                log.warn("================当前规则为:"+syncRule.getSyncRuleName());
                List<ScanMetadata> scanMetadataList = scanMetadataService.qyScanMetadataTable(syncRuleDTO);
                scanMetadataList.forEach(s ->{
                    log.warn("==========当前表为:"+s.getTblName());
                    if("源标识同步".equals(syncRule.getSyncMode())&&"分区".equals(syncRule.getIsPart())&&"无时间分区".equals(syncRule.getIsTimePart())){
                        //拼接条件
                        String where = syncRuleConfServiceImpl.whereLike(syncRule);
                        //查询该表源的hdfs路径
                        List<String> hdfsList = hiveToOracleService.selectHiveLocation(where,s.getTblName(),s.getTblDatabaseName());
                        if(hdfsList!=null&&hdfsList.size()>0){
                            for (String hdfs: hdfsList) {
                                HdfsHeatBackup kup = new HdfsHeatBackup();
                                WhereScriptDTO whereScriptDTO = new WhereScriptDTO();
                                String lo = hdfs.substring(syncRuleConfServiceImpl.getSlashIndex(hdfs,6) + 1);
                                kup.setSourceAddress(wGCluster.getHdfsPrefixPath()+lo);
                                kup.setTargetAddress(uatCluster.getHdfsPrefixPath()+prefix+lo);
                                kup.setSourceDbName(s.getTblDatabaseName());
                                kup.setTargetDbName(prefix+s.getTblDatabaseName());
                                kup.setSourceTableName(s.getTblName());
                                kup.setTargetTableName(s.getTblName());
                                kup.setBackupSort("1");
                                ruleToDistCPDTO.setBackupSort("1");
                                whereScriptDTO.setSourceDbName(s.getTblDatabaseName());
                                whereScriptDTO.setSourceTableName(s.getTblName());
                                kup.setSourceData(syncRule.getSourceName());
                                ruleToDistCPDTO.setSourceDataName(syncRule.getSourceName());
                                String[] partNames = hdfs.substring(syncRuleConfServiceImpl.getSlashIndex(hdfs,8) + 1).split("=");
                                whereScriptDTO.setSourceQueryCriteria(StrUtil.replace(StrUtil.replace(hdfs.substring(syncRuleConfServiceImpl.getSlashIndex(hdfs,8) + 1),"=","='"),partNames[0],"cast( "+ partNames[0] +" as string )") + "'");
                                whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
                                whereScriptDTO.setTargetTableName(s.getTblName());
                                whereScriptDTO.setTargetDbName(prefix+s.getTblDatabaseName());
                                whereScriptDTOS.add(whereScriptDTO);
                                hdfsHeatBackups.add(kup);
                                log.warn("==========================【全量源标识】当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
                            }
                        }
                    }else {
                        WhereScriptDTO whereScriptDTO = new WhereScriptDTO();
                        HdfsHeatBackup kup = new HdfsHeatBackup();
                        if("分区".equals(syncRule.getIsPart())&&"特殊表含时间分区".equals(syncRule.getIsTimePart())){
                            kup.setSourceAddress(wGCluster.getHdfsPrefixPath()+s.getTblDatabaseName()+".db/"+s.getTblName()+"/"+syncRule.getPartValueWhite());
                            kup.setTargetAddress(uatCluster.getHdfsPrefixPath()+prefix+s.getTblDatabaseName()+".db/"+s.getTblName()+"/"+syncRule.getPartValueWhite());
                            String[] partNames = syncRule.getPartValueWhite().split("=");
                            //当有时间分区时对比条件拼接要加上时间分区
                            if(StrUtil.isBlank(syncRule.getPartNameWhite())){
                                whereScriptDTO.setSourceQueryCriteria(StrUtil.replace(StrUtil.replace(syncRule.getPartValueWhite(),"=","='"),partNames[0],"cast( "+ partNames[0] +" as string )") + "'");
                            }else{
                                LambdaQueryWrapper<TimeFieldMatching> timeFieldWrapper = Wrappers.lambdaQuery();
                                timeFieldWrapper.select(TimeFieldMatching::getTimeFieldKay, TimeFieldMatching::getTimeFieldValue);
                                timeFieldWrapper.eq(TimeFieldMatching::getTimeFieldKay, syncRule.getPartNameWhite())
                                        .eq(TimeFieldMatching::getDel, CommonConstants.STATUS_NORMAL);
                                TimeFieldMatching matching = timeFieldMatchingService.getOne(timeFieldWrapper);
//                                String endTime = DateUtil.format(DateUtil.date(), matching.getTimeFieldValue());
//                                int betweenDay = -750;
//                                String staTime = DateUtil.format(DateUtil.offset(DateUtil.date(), DateField.DAY_OF_MONTH, betweenDay), matching.getTimeFieldValue());
                                String endTime = String.valueOf(dto.getEndDate());
                                String staTime = String.valueOf(dto.getStaDate());
                                whereScriptDTO.setSourceQueryCriteria(StrUtil.replace(StrUtil.replace(syncRule.getPartValueWhite(),"=","='"),partNames[0],"cast( "+ partNames[0] +" as string )") + "' and "+"cast( "+ syncRule.getPartNameWhite() +" as int )"+ " BETWEEN " + staTime +" and "+ endTime);
                            }
                            whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
                        }else {
                            kup.setSourceAddress(wGCluster.getHdfsPrefixPath()+s.getTblDatabaseName()+".db/"+s.getTblName());
                            kup.setTargetAddress(uatCluster.getHdfsPrefixPath()+prefix+s.getTblDatabaseName()+".db/"+s.getTblName());
                        }
                        kup.setSourceDbName(s.getTblDatabaseName());
                        kup.setTargetDbName(prefix+s.getTblDatabaseName());
                        kup.setSourceTableName(s.getTblName());
                        kup.setTargetTableName(s.getTblName());
                        kup.setBackupSort("1");
                        ruleToDistCPDTO.setBackupSort("1");
                        whereScriptDTO.setSourceDbName(s.getTblDatabaseName());
                        whereScriptDTO.setSourceTableName(s.getTblName());
                        if(StrUtil.isNotBlank(syncRule.getSourceName())){
                            kup.setSourceData(syncRule.getSourceName());
                            ruleToDistCPDTO.setSourceDataName(syncRule.getSourceName());
                        }else{
                            kup.setSourceData("无源");
                            ruleToDistCPDTO.setSourceDataName("无源");
                        }
                        if(StrUtil.isBlank(whereScriptDTO.getSourceQueryCriteria())){
                            whereScriptDTO.setSourceQueryCriteria("");
                            whereScriptDTO.setTargetQueryCriteria(whereScriptDTO.getSourceQueryCriteria());
                        }
                        whereScriptDTO.setTargetTableName(s.getTblName());
                        whereScriptDTO.setTargetDbName(prefix+s.getTblDatabaseName());
                        whereScriptDTOS.add(whereScriptDTO);
                        log.warn("==========================当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
                        hdfsHeatBackups.add(kup);
                    }
                });
                //近2年/近几年数据一定包含时间分区
            }else if(syncRule.getSyncRange()>0){
                int betweenDay = syncRule.getSyncRange();
                log.warn("================当前规则为:"+syncRule.getSyncRuleName());
                //分组拿表名
                List<ScanMetadata> scanMetadataList = scanMetadataService.qyScanMetadataTable(syncRuleDTO);
                scanMetadataList.forEach(s ->{
                    log.warn("==========当前表为:"+s.getTblName());
                    //查询该表的全部分区字段
                    List<String> hdfsList = hiveToOracleService.selectHivePkeyName(s.getTblName(),s.getTblDatabaseName());
                    //循环匹配分区字段配置表 拿到字段名称和字段格式
                    for (String h : hdfsList) {
                        LambdaQueryWrapper<TimeFieldMatching> timeFieldWrapper = Wrappers.lambdaQuery();
                        timeFieldWrapper.select(TimeFieldMatching::getTimeFieldKay, TimeFieldMatching::getTimeFieldValue);
                        timeFieldWrapper.eq(TimeFieldMatching::getTimeFieldKay, h).eq(TimeFieldMatching::getDel, CommonConstants.STATUS_NORMAL);
                        TimeFieldMatching matching = timeFieldMatchingService.getOne(timeFieldWrapper);
                        //不为空就匹配成功
                        if (ObjectUtils.isNotEmpty(matching)) {
                            //拼接条件
                            String where = syncRuleConfServiceImpl.whereLike(syncRule);
                            //获取当前规则源时间分区内的hdfs地址
                            List<String> locationList = hiveToOracleService.selectHiveTableHDFSBee(s.getTblName(), where, matching.getTimeFieldKay(), String.valueOf(dto.getStaDate()),String.valueOf(dto.getEndDate()), matching.getTimeFieldValue(), s.getTblDatabaseName());
                            if(locationList!=null&&locationList.size()>0){
                                //拼接数据对比所需条件 出现类似于订单TA源分区多种value值
                                WhereScriptDTO whereScriptDTO =  syncRuleConfServiceImpl.whereScript(locationList, s.getTblDatabaseName(), s.getTblName(), matching, syncRule.getSyncMode(), -betweenDay, prefix);
                                Set<kvDTO> locationGorup = new HashSet<>();
                                locationList.forEach(lo -> {
                                    HdfsHeatBackup kup = new HdfsHeatBackup();
                                    //切割分区信息
                                    String partName = "";
                                    if (StrUtil.split(lo, s.getTblName()).length == 2) {
                                        partName = StrUtil.split(lo, s.getTblName())[1];
                                    } else {
                                        log.warn("===========" + lo + "===========");
                                    }
                                    //（为了避免该表有多个时间分区字段） 且不等于无源
                                    if(StrUtil.isNotBlank(partName)&&StrUtil.isNotBlank(syncRule.getSourceName())){
                                        kvDTO kvDTO = new kvDTO();
                                        //判断时间分区在前在后 切割源标识
                                        String[] names = partName.substring(1,partName.length()).split("/");
                                        String[] partNamesQ = names[0].split("=");
                                        if(names.length>=2){
                                            String[] partNamesH = names[1].split("=");
                                            //分区在前
                                            if(matching.getTimeFieldKay().equals(partNamesQ[0])){
                                                kvDTO.setSourceKey(names[1]);
                                                kvDTO.setSourceValue(partNamesH[0]);
                                                kvDTO.setTimeKey(partNamesQ[0]);
                                                locationGorup.add(kvDTO);
                                                //分区在后
                                            }else if(matching.getTimeFieldKay().equals(partNamesH[0])){
                                                kvDTO.setSourceKey(names[0]);
                                                kvDTO.setSourceValue(partNamesQ[0]);
                                                kvDTO.setTimeKey(partNamesH[0]);
                                                locationGorup.add(kvDTO);
                                            }
                                        }else{
                                            kvDTO.setSourceKey(names[0]);
                                            locationGorup.add(kvDTO);
                                        }
                                    }
                                    lo = wGCluster.getHdfsPrefixPath() + s.getTblDatabaseName() + ".db/" + s.getTblName() + partName;
                                    kup.setSourceAddress(lo);
                                    kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo, wGCluster.getHdfsPrefixPath(), uatCluster.getHdfsPrefixPath()), "/" + s.getTblDatabaseName() + ".db/", "/" + prefix + s.getTblDatabaseName() + ".db/"));
                                    kup.setSourceDbName(s.getTblDatabaseName());
                                    kup.setTargetDbName(prefix + s.getTblDatabaseName());
                                    kup.setSourceTableName(s.getTblName());
                                    kup.setTargetTableName(s.getTblName());
                                    kup.setPkeyName(matching.getTimeFieldKay());
                                    kup.setSyncRange(syncRule.getSyncRange());
                                    kup.setBackupSort("1");
                                    ruleToDistCPDTO.setBackupSort("1");
                                    if (StrUtil.isNotBlank(syncRule.getSourceName())) {
                                        kup.setSourceData (syncRule.getSourceName());
                                        ruleToDistCPDTO.setSourceDataName(syncRule.getSourceName());
                                    } else {
                                        kup.setSourceData("无源");
                                        ruleToDistCPDTO.setSourceDataName("无源");
                                    }
                                    hdfsHeatBackups.add(kup);
                                });
                                //源分区单种value值（为了避免该表有多个时间分区字段） 或者无源
                                if((locationGorup!=null&&locationGorup.size()==1)||StrUtil.isBlank(syncRule.getSourceName())){
                                    log.warn("==========================当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
                                    whereScriptDTOS.add(whereScriptDTO);
                                    break;
                                    //源分区多种value值
                                }else if(locationGorup!=null&&locationGorup.size()>1){
                                    for (kvDTO kv : locationGorup) {
                                        WhereScriptDTO wsDTO = new WhereScriptDTO();
//                                    String endTime = DateUtil.format(DateUtil.date(), matching.getTimeFieldValue());
//                                    String staTime =  DateUtil.format(DateUtil.offset(DateUtil.date(), DateField.DAY_OF_MONTH, -betweenDay), matching.getTimeFieldValue());
                                        String endTime = String.valueOf(dto.getEndDate());
                                        String staTime = String.valueOf(dto.getStaDate());
                                        String whereSql = StrUtil.replace(StrUtil.replace(kv.getSourceKey(),"="," = '"),kv.getSourceValue(),"cast( "+ kv.getSourceValue() +" as string )") + "' and "+"cast( "+ kv.getTimeKey() +" as int )"+ " BETWEEN " + staTime +" and "+ endTime;
                                        wsDTO.setSourceDbName(s.getTblDatabaseName());
                                        wsDTO.setSourceTableName(s.getTblName());
                                        wsDTO.setTargetDbName(prefix+s.getTblDatabaseName());
                                        wsDTO.setTargetTableName(s.getTblName());
                                        wsDTO.setSourceQueryCriteria(whereSql);
                                        wsDTO.setTargetQueryCriteria(wsDTO.getSourceQueryCriteria());
                                        log.warn("==========================当前数据比对条件"+whereScriptDTO.getSourceQueryCriteria());
                                        whereScriptDTOS.add(wsDTO);
                                    }
                                }
                            }
                        }
                    }
                });
            }
        });
        //将比对sql对象 赋值给传输DTO
        ruleToDistCPDTO.setWhereScriptDTOS(whereScriptDTOS);
        Set<String> set = new HashSet<>();
        for (HdfsHeatBackup hdfs: hdfsHeatBackups) {
            set.add(hdfs.getSourceName());
        }
        for (String name : set) {
            if(StrUtil.isNotBlank(uatSyncLogsTask.getLibraryTableName())){
                uatSyncLogsTask.setLibraryTableName(uatSyncLogsTask.getLibraryTableName()+"\n"+name);
            }else{
                uatSyncLogsTask.setLibraryTableName(name);
            }
        }
        ruleToDistCPDTO.setMessageId(mess.getId());
        ruleToDistCPDTO.setHdfsHeatBackups(hdfsHeatBackups);
        uatSyncLogsTask.setTableHdfsNumber(""+hdfsHeatBackups.size());
        uatSyncLogsTaskService.saveOrUpdate(uatSyncLogsTask);
        ruleToDistCPDTO.setUatSyncLogsTask(uatSyncLogsTask);
        String s  = "";
        //不为空且的等于“是”
        if(StrUtil.isNotBlank(dto.getCpWhether())&&"是".equals(dto.getCpWhether())){
            UatSyncLogs uatSyncLogs = new UatSyncLogs();
            if(ObjectUtil.isNotEmpty(ruleToDistCPDTO.getHdfsHeatBackups())&&ruleToDistCPDTO.getHdfsHeatBackups().size() > 0) {
                s = ruleToDistCPService.heatBackupWgToUat(ruleToDistCPDTO);
            }else{
                //当前结束时间
                Date endDate = DateUtil.date();
                uatSyncLogs.setTaskId(uatSyncLogsTask.getId());
                uatSyncLogs.setTaskName(uatSyncLogsTask.getTaskName());
                uatSyncLogs.setStandbyTime(DateUtil.formatBetween(staDate,endDate, BetweenFormater.Level.SECOND));
                uatSyncLogs.setLog("["+DateUtil.now()+"]【暂无HDFS路径信息】");
                uatSyncLogs.setTaskState(ScanCommonConstants.ClusterName.YC.value);
                uatSyncLogsService.save(uatSyncLogs);
                log.warn("【暂无HDFS路径信息】");
            }
        }
        //当前结束时间
        Date endDate = DateUtil.date();
        uatSyncLogsTask.setStandbyTime(DateUtil.formatBetween(staDate,endDate, BetweenFormater.Level.SECOND));
        if(StrUtil.isNotBlank(s)&&"成功".equals(s)){
            uatSyncLogsTask.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
            uatSyncLogsTaskService.updateById(uatSyncLogsTask);
            mess.setDataType(ScanCommonConstants.ClusterName.YWC.value);
            messageCenterService.updateById(mess);
            return true;
        }else{
            uatSyncLogsTask.setTaskState(ScanCommonConstants.ClusterName.YC.value);
            uatSyncLogsTaskService.updateById(uatSyncLogsTask);
            mess.setDataType(ScanCommonConstants.ClusterName.YC.value);
            messageCenterService.updateById(mess);
            return false;
        }
    }
}
